package com.beta.mq.redis.consumer.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.beta.mq.redis.RedisMqProperties;
import com.beta.mq.redis.consumer.Consumer;
import com.beta.mq.redis.model.Message;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * 基于Redis无界队列实现的MQ存储中心的消费者
 */
@Slf4j
public class RedisBlockingQueueConsumer implements Consumer {
    private RedisMqProperties properties;
    private RedissonClient redissonClient;

    public RedisBlockingQueueConsumer(RedissonClient redissonClient, RedisMqProperties properties) {
        this.redissonClient = redissonClient;
        this.properties = properties;
    }

    @Override
    public Message getMessage(String topic) throws InterruptedException {
        String msg = (String)redissonClient.getBlockingQueue(topic).take();
        return JSON.parseObject(msg,Message.class);
    }

    @Override
    public List<Message> getBatchMessage(String topic,Integer limit) throws InterruptedException {
        if (limit == null || limit <= 0 || limit > 10000){
            limit = 1000;
        }
        List msgs = redissonClient.getBlockingQueue(topic).poll(limit);
        List<Message> result = new ArrayList<>();
        //获取到的数据不为空
        if (!CollectionUtils.isEmpty(msgs)){
            for (Object msg : msgs) {
                //如果JSON转换失败则跳过
                try {
                    if (!StringUtils.isEmpty(msg)){
                        Message message = JSON.parseObject(msg.toString(),Message.class);
                        result.add(message);
                    }
                }catch (Exception ex){
                    log.warn("getBatchMessage warn,message is [{}]",msg);
                }
            }
        }
        return result;
    }

    @Override
    public void retry(String topic, Message message) throws InterruptedException {
        //如果消息执行失败，重试
        if (properties.getTryTimes() == 0 || properties.getTryTimes() < -1) {
            return;//不重试
        }
        if (properties.getTryTimes() == -1 || message.getTryTimes() < properties.getTryTimes()) {
            message.setTryTimes(message.getTryTimes() + 1);
            redissonClient.getBlockingQueue(topic).put(message);
        }
    }

    @Override
    public void retryBatch(String topic, List<Message> messages) throws InterruptedException {
        //如果消息执行失败，重试
        if (properties.getTryTimes() == 0 || properties.getTryTimes() < -1) {
            return;//不重试
        }
        for (Message message : messages) {
            if (properties.getTryTimes() == -1 || message.getTryTimes() < properties.getTryTimes()) {
                message.setTryTimes(message.getTryTimes() + 1);
                redissonClient.getBlockingQueue(topic).put(message);
            }
        }

    }
}
